package com.minelx.parallel;

import com.minelx.taskscheduler.CallbackTask;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Supplier;

public class Job<START, THEN, END> {

	private final Mapper<START, THEN> mapper;

	private final Reducer<THEN, END> reducer;

	private Job(Mapper<START, THEN> mapper,
	            Reducer<THEN, END> reducer) {
		this.mapper = mapper;
		this.reducer = reducer;
	}

	public END work(Iterator<START> content) {
		content.forEachRemaining(mapper::map);
		// it's done, all content have been accumulated.
		return reducer.reduce();
	}

	public Supplier<END> workParallel(Iterator<START> content, int nThreads) {
		ExecutorService service = Executors.newFixedThreadPool(nThreads);
		CallbackTask tasks = new CallbackTask(nThreads);
		Executor<START> executor = new Executor<>(content, mapper::map, tasks);
		// run all threads
		for (int at = 0; at < nThreads; at++) {
			service.submit(executor);
		}
		// shutdown at the end of execution
		service.shutdown();
		return () -> reduceAfterTasksCompleted(tasks);
	}

	private END reduceAfterTasksCompleted(CallbackTask task) {
		try {
			task.waitForCompletion();
		} catch (InterruptedException e) {
			throw new RuntimeException("interrupted while doing tasks, Caused By: ", e);
		}
		return reducer.reduce();
	}

	public static class Builder<START, THEN, END> {

		private Function<START, THEN> mapFunction;

		private Function<List<THEN>, END> reduceFunction;

		public Builder<START, THEN, END> map(Function<START, THEN> mapFunction) {
			this.mapFunction = mapFunction;
			return this;
		}

		public Builder<START, THEN, END> reduce(Function<List<THEN>, END> reduceFunction) {
			this.reduceFunction = reduceFunction;
			return this;
		}

		public Job<START, THEN, END> build() {
			Reducer<THEN, END> reducer = new Reducer<>(reduceFunction);
			Mapper<START, THEN> mapper = new Mapper<>(reducer, mapFunction);
			return new Job<>(mapper, reducer);
		}
	}
}

























